
#include "BigQ.h"

//Array to maintain number of pages in each run to check for record on next page error
vector<int> pages_in_run;

//External sort object where all the sorting takes place
ExternalSort sortIt;

//Worker thread
pthread_t worker;

//Worker thread function
void *work(void *arg)
{
	sortIt.RunGeneration();
	sortIt.RunMerging();

}

//Constructor for bigQ class
BigQ :: BigQ(Pipe &in, Pipe &out, OrderMaker &sortorder, int runlen)
{

	sortIt.tempOrderMaker=&sortorder;
	sortIt.inputpipe=&in;
	sortIt.outputpipe=&out;
	sortIt.passedrunlen=runlen;
	pthread_create(&worker,NULL,work,NULL);
	cout<<"thread crted"<<endl;

}

//Destructor for bigQ class
BigQ :: ~BigQ(){

}

// To check how many times we execute comparing vector for run Generation comparison
//Function for comparing vector in Phase 1 or Run Generation
bool ExternalSort :: comparingvector(Record* left,Record* right)
{
	ComparisonEngine compEng;
	int res;

	//tempOrderMaker is in passed from sortIt(External Sort) object which is local for this class
	res=compEng.Compare(left,right,sortIt.tempOrderMaker);

	if(res<0)
		return true;
	else
		return false;

}

/*
 *Combination made for keep this in priority queue.
 *Combination made to keep run number with record so that we can know from which run record was deleted, so that we can add record from the same run
 *
 */
class RecordRunCombination{

public:

	Record* recordFromRun;
	int runNumberOfRecord;

	//Shallow copy used for now  need to check if need to do deep copy
	RecordRunCombination(Record* record, int runNumber)
	{
		recordFromRun = record;
		runNumberOfRecord = runNumber;
	}

	//***Need to check if need to delete memory allocated if we do deep copy(Problem maybe due to memory allocation here)
	virtual ~RecordRunCombination(){}

};

/*
 * Class to compare records from priority queue to be used in run merging phase
 *
 *
 * */
class PairComparator
{

public:

	//overloading operator function for comparison of Priority queue
	bool operator()(RecordRunCombination* left,RecordRunCombination* right)
	{
		ComparisonEngine compEng;
		int res;

		//fetching record frm struct and then sending it to compare records to comparison engine class
		res=compEng.Compare(left->recordFromRun,right->recordFromRun,sortIt.tempOrderMaker);

		if(res<0)
			return false;
		else
			return true;

	}

};

//External sort constructor
//**Need to check if need to initialize something here to streamline the code
ExternalSort::ExternalSort()
{

}

//Run generation function
void ExternalSort::RunGeneration()
{
	//Initializing runs and counters
	int currRunNumber=0;
	totalRuns=0;

	//**Open the file(sorted_File) to be written
	myFile.Open(0,"sorted_File.txt");

	//Counter to count if page number is less then given run length
	int pagecount=0;

	Record* recordRead,*tempRecord;
	recordRead= new Record;

	//-1 cuz we want first page to be page number 0
	pages_in_run.push_back(-1);

	//Removing records or getting records from input Pipe
	while((inputpipe->Remove(recordRead)))
	{
		tempRecord= new Record;
		tempRecord->Copy(recordRead);

		//Check if bufferPage is full, if it is full then dump it into the file and increment pageCount for that run
		if(!bufferPage.Append(recordRead))
		{
			pagecount++;

			//To check if it is last page of the current run
			if(pagecount==passedrunlen)
			{
				bufferPage.EmptyItOut();
				pagecount=0;

				//Creating new element in vector array for next new run
				pages_in_run.push_back(-1);

				//Sort the current run
				sort(sortingArray.begin(),sortingArray.end(),ExternalSort::comparingvector);

				//Dump the page to the file
				dumpOnDisk();
				totalRuns++;

				//cout<<"Run completed...next run started\n ";
				//Clear vector containing records of the previous vector array
				sortingArray.clear();

				//since Record wasn't appended so we have to append it now.
				bufferPage.Append(recordRead);
			}
			//Means this is not the last page of the current run ....somewhere in the middle of any run but page is full
			else
			{
				bufferPage.EmptyItOut();
				bufferPage.Append(recordRead);
			}
		}//Else buffer page is not full and push the record into the vector array
		sortingArray.push_back(tempRecord);
		//delete tempRecord;
	}

	//cout<<"size  of vector : "<<sortingArray.size()<<endl;
	bufferPage.EmptyItOut();

	//If pipe is empty but wwe still have records in array
	if(!(sortingArray.size()==0)){

		//Sort the run
		sort(sortingArray.begin(),sortingArray.end(),ExternalSort::comparingvector);

		//Dump the remaining records into the file
		dumpOnDisk();
		++totalRuns;
		cout<<"When pipe is emptied and we still have records sort them***********\n Total Runs is "<<totalRuns<<endl;

		//Clear vector
		sortingArray.clear();
	}

	//************For error checking if needed in future
	/*for(int i=0;i<pages_in_run.size();i++)
		cout<<"Run number "<<i<<"Pages "<<pages_in_run[i]<<endl;*/
	//cout<<"RunGeneration : Total number of runs genrated are........... "<<totalRuns<<endl;
	cout<<"Going out of run generation\n";
	delete recordRead;

}

//Function to dump records from page on to the file(sorted_File)
void ExternalSort::dumpOnDisk()
{
	Record* localrecord;
	localrecord=new Record;

	//cout<<"In dump disk\n";

	//Loop to dump each record from sorting array into the file
	for(int i=0;i<sortingArray.size();i++)
	{
		localrecord->Copy(sortingArray[i]);
		if(!bufferPage.Append(localrecord))
		{
			myFile.AddPage(&bufferPage,page_in_file);
			pages_in_run[totalRuns]++;

			//**Debugging information
			//cout<<"Run "<<totalRuns<<" Page "<<pages_in_run[totalRuns]<<"Records "<<bufferPage.numRecs<<endl;

			bufferPage.EmptyItOut();
			bufferPage.Append(localrecord);
			page_in_file++;

		}

	}

	//since last page wasn't full so adding that incomplete page on completion of run
	myFile.AddPage(&bufferPage,page_in_file);
	pages_in_run[totalRuns]++;

	//**Debugging information
	//cout<<"Run "<<totalRuns<<" Page "<<pages_in_run[totalRuns]<<"Records "<<bufferPage.numRecs<<endl;
	bufferPage.EmptyItOut();
	page_in_file++;
	delete localrecord;
}

//Function for calculation of offset primarily used to calculate the location of 1st page of current run in the file.
int ExternalSort ::GetPageOffset()
{
	int runCounter=0;
	int tempPageOffset=0;

	//Will successively add pages in run every time and add it
	while(runCounter<runIndex)
	{
		tempPageOffset=pages_in_run[runCounter]+1+tempPageOffset;
		runCounter++;
	}

	return tempPageOffset;

}

//Function to do Run merging
void ExternalSort::RunMerging(){
	//***********************************PHASE_2*********Run Merge****************
	Record* tempRecord;
	int pageOffset;
	RecordRunCombination* recordRunStructure;

	//Array to maintain counter of current page of a run that is being read
	int currPageIndexOfRun[totalRuns];

	Page pageOfRun[totalRuns];
	//Setting run and page indexes
	runIndex = 0;
	pageOffset = 0;

	//Set all currPageIndexOfRun = 0
	for(int i = 0; i<totalRuns; i++){
		currPageIndexOfRun[i]=0;
	}

	//pqueue contains the structure of record and run number of that record and comparison class as parameters
	priority_queue <RecordRunCombination*,vector<RecordRunCombination*>,PairComparator> pQueue;

	//Set tempRecord to null to remove any previous memory records
	tempRecord = NULL;

	//************ Initializing runs for merging 1st time reading first page of all run and building initial priority queue
	while(runIndex<totalRuns)
	{
		//Getting page offset for the runs
		pageOffset=GetPageOffset();

		//**Debugging information
		//cout<<"Fetching Page number "<<pageOffset<<"of run "<<runIndex<<endl;

		//Fetching 1st page of every run.
		myFile.GetPage(&pageOfRun[runIndex],pageOffset);

		//Getting first record of the run
		tempRecord = new Record;
		pageOfRun[runIndex].GetFirst(tempRecord);
		recordRunStructure = new RecordRunCombination(tempRecord,runIndex);

		//Adding 1st record of the run and its run number as a Record Run Combination object to the priority queue
		pQueue.push(recordRunStructure);

		//Deleting record's reference to get new reference in new memory location
		tempRecord = NULL;
		delete tempRecord;
		recordRunStructure = NULL;
		delete recordRunStructure;
		runIndex++;

	}

	//*******************Starting popping top of pQueue into output pipe
	pageOffset=0;
	int outputpipecheckcount=0;

	//repeat until queue is empty
	while(!pQueue.empty())
	{
		tempRecord=new Record();

		//runIndex = pQueue.top.second()
		runIndex=pQueue.top()->runNumberOfRecord;

		//**********Debugging information
		//cout<<"*********************PageOffset "<<pageOffset<<" Run number - "<<runIndex<<"Page number of run - "<<currPageIndexOfRun[runIndex]<<endl;


		//remove top of priority queue and insert it into output pipe.
		Record* insertIntoPipe = new Record;
		insertIntoPipe = pQueue.top()->recordFromRun;
		outputpipecheckcount++;
		outputpipe->Insert(insertIntoPipe);
		insertIntoPipe = NULL;
		delete insertIntoPipe;

		//Remove that element from the queue
		pQueue.pop();

		//*********Inserting next record into pQueue and removing top continously
		//fetching first record from temporary run page(runIndex)
		//if GetFirst returns 0 that means page is out of records so fetch next page
		if(pageOfRun[runIndex].GetFirst(tempRecord)==0)
		{
			//Increment currentPageIndexOfRun for that run
			++currPageIndexOfRun[runIndex];

			//Set offset to get records from the next page of the run
			pageOffset = GetPageOffset()+currPageIndexOfRun[runIndex];

			//Checking that this page is less then run length and number of pages doesn't cross total number of pages in file.
			if ((currPageIndexOfRun[runIndex]<=pages_in_run[runIndex]) && ( pageOffset < myFile.GetLength()-1))
			{
				myFile.GetPage(&pageOfRun[runIndex],pageOffset);

				//Fetch first record from newly inserted buffer page(GetFirst)
				pageOfRun[runIndex].GetFirst(tempRecord);

				//push temporary record temprecord from temporary run as a record run combination structure into the queue.
				recordRunStructure = new RecordRunCombination(tempRecord,runIndex);
				pQueue.push(recordRunStructure);

			}

		}else{/*Means that the current page of the run still has records so push that record into the queue*/

			recordRunStructure = new RecordRunCombination(tempRecord,runIndex);
			pQueue.push(recordRunStructure);

		}

		//Deleting memory used
		tempRecord=NULL;
		delete tempRecord;
		recordRunStructure = NULL;
		delete recordRunStructure;


		//******************Queue become empty and all elements are in output pipe :)



	}

	//*******************Out Pipe testing******************
	//cout<<"File length "<<myFile.GetLength()<<endl;



	//testingRecord = new Record;
	//cout<<"***************Testing run merging ************"<<endl;
	//while pipe is not empty
	//int finalwhilecount=0;
	/*while(outputpipe->Remove(testingRecord)){
	//	cout<<"********Inside testing while***********"<<endl;
	//out.Remove(testingRecord);
	//***************testingRecord->Print(mySchema);
	finalwhilecount++;

	//out.Remove(testingRecord);
	//testingRecord->Print(mySchema);
	//out.Remove(testingRecord);
	//testingRecord->Print(mySchema);
	//out.Remove(testingRecord);
	//testingRecord->Print(mySchema);

	//testingRecord = NULL;
	//delete testingRecord;
	//testingRecord = new Record;
	}*/

	//cout<<"Records compared of struct recordruncomb is "<<finalwhilecount<<endl;
	//cout<<"Vector count check"<<vectorcountcheck<<endl;
//************Out pipe testing information ends********

	//Close the file and shut down the pipe to unlock the thread being used by the pipe
	myFile.Close();
	outputpipe->ShutDown ();

}

//**Check which objects need to be destroyed to free the memory
//External Sort destructor
ExternalSort::~ExternalSort() {
	//delete outputpipe;
	//delete tempOrderMaker;
	// TODO Auto-generated destructor stub
	//delete recordRead;
	//delete tempRecord;
	//delete myFile;
	//delete bufferPage;
	//delete tempOrderMaker;
}
